package cn.edu.nju.ws.sview.crawler;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.PriorityQueue;

import cn.edu.nju.ws.sview.rdf.URIResource;


/**
 * 任务调度进程
 * @author Gong Saisai
 *
 */
public class Scheduler implements Runnable{
	 private Flag<Boolean> stopFlag ;
	 private LinkedList<URIResource> pool = null;
	 private int maxThreadNum;
	 private  Object lockOfURIStatus = null;
	 private  HashSet<URIResource> done = null;
 	 private  HashSet<URIResource> busy = null;
 	 private  PriorityQueue<Element> waiting = null;
 	 private CrawlerMeta meta= null;
 	//加快查询waiting优先队列元素的速度
 	 private  HashMap<URIResource,Element> resource2Elements = null;
 	 
	 public Scheduler(HashSet<URIResource> done, HashSet<URIResource> busy,PriorityQueue<Element> waiting,
			 HashMap<URIResource,Element> resource2Elements,Object lockOfURIStatus,int thread_num,Flag<Boolean> stopFlag,CrawlerMeta meta){
		this.done = done;
		this.busy = busy;
		this.waiting = waiting;
		this.resource2Elements =resource2Elements;
		this.lockOfURIStatus = lockOfURIStatus;
		this.maxThreadNum = thread_num;
		this.stopFlag = stopFlag;
		this.meta = meta;
		this.pool = new LinkedList<URIResource>();
	 }
	
	
	@Override
	public void run() {
		// TODO Auto-generated method stub
		ArrayList<Thread> threads = new ArrayList<Thread>(this.maxThreadNum);
		for (int i = 0; i < this.maxThreadNum; i++) {
			Thread thread = new Thread(new TaskThread(this.pool,this.done,this.busy,this.lockOfURIStatus,this.meta));
			threads.add(thread) ;
			thread.start();
		}
		
		while (true) {
			URIResource uriResource = null;
			boolean waitAndContinue = false;
			synchronized (lockOfURIStatus) {
				if (this.waiting.isEmpty()) {
					waitAndContinue = true;
				}else{
					uriResource = this.waiting.poll().getResource();
					this.resource2Elements.remove(uriResource) ;
					this.busy.add(uriResource);
				}
			}
			if(waitAndContinue){
				try {
					Thread.sleep(10) ;
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				continue;
			}
			if(uriResource ==null){
				for(int i=0;i<this.maxThreadNum;i++){
					synchronized(this.pool){
						this.pool.offer(null) ;
					}
				}
				break;
			}
			
			synchronized(this.pool){
				this.pool.offer(uriResource) ;
			}
		}
		
		while(true){
			boolean finish = true;
			for(int i=0;i<this.maxThreadNum;i++){
				if(threads.get(i)!=null&&threads.get(i).isAlive()){
					finish =false;
					break;
				}
			}
			if(finish)
				break;
			try {
				Thread.sleep(10) ;
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		
		this.pool.clear();
		threads.clear();
		this.stopFlag.setFlag(new Boolean(true));
	}

}
